#include "myRpcChannel.h"

#include "rpcHeader.pb.h"

#include <arpa/inet.h>
#include <cerrno>
#include <cstdint>
#include <cstdio>
#include <glog/logging.h>
#include <google/protobuf/descriptor.h>
#include <google/protobuf/io/coded_stream.h>
#include <google/protobuf/io/zero_copy_stream_impl_lite.h>
#include <netinet/in.h>
#include <sstream>
#include <sys/socket.h>
#include <unistd.h>
#include <vector>
namespace RPC
{

    MyrpcChannel::MyrpcChannel( std::string ip, short port, bool connectNow )
        : m_client_fd( -1 )
        , m_ip( ip )
        , m_port( port )
    {
        if ( !connectNow )
        {
            // 这里就是允许延迟连接
            return;
        }
        std::string errMsg;
        bool ret = newConnect( ip.c_str(), port, errMsg );
        // 连接失败时，可以重新连接5次
        int tryConnect = 5;
        while ( !ret && tryConnect > 0 )
        {
            tryConnect--;
            std::cerr << errMsg << std::endl;
            LOG( ERROR ) << errMsg;
            ret = newConnect( ip.c_str(), port, errMsg );
        }
    }

    void MyrpcChannel::CallMethod( const google::protobuf::MethodDescriptor *method,
                                   google::protobuf::RpcController *controller,
                                   const google::protobuf::Message *request,
                                   google::protobuf::Message *response,
                                   google::protobuf::Closure *done )
    {
        // 首先从MethodDescriptor中获得方法的相关数据
        if ( m_client_fd == -1 )
        {
            std::string errMsg;
            // 进行重新连接
            bool ret = newConnect( m_ip.c_str(), m_port, errMsg );
            if ( !ret )
            {
                std::cerr << "[func-MprpcChannel::CallMethod]重连接ip：" << m_ip << "port：" << m_port << "失败" << std::endl;
                LOG( ERROR ) << "[func-MprpcChannel::CallMethod]重连接ip：" << m_ip << "port：" << m_port << "失败";
                controller->SetFailed( errMsg );
                return;
            }
            else
            {
                std::cerr << "[func-MprpcChannel::CallMethod]重连接ip：" << m_ip << "port：" << m_port << "重新连接成功" << std::endl;
                LOG( INFO ) << "[func-MprpcChannel::CallMethod]重连接ip：" << m_ip << "port：" << m_port << "重新连接成功";
            }
        }

        // 从method获取rpc服务描述符
        const google::protobuf::ServiceDescriptor *sd = method->service();
        // 获取服务名
        const std::string service_name = sd->name();

        // 获取方法名
        const std::string method_name = method->name();

        // 获取参数的序列化字符串长度args_size, request 里面封装的是调用服务方法时传递的参数数据。这些数据自动序列化成了Message对象了；
        // 比如我们调用选举的方法，这个方法里面会封装任期等参数，这些参数就是需要使用到的数据，服务名+方法名+需要使用到的数据，就完成了一个远程调用
        uint32_t args_size = 0;
        std::string args_str;
        if ( request->SerializePartialToString( &args_str ) )
        {
            args_size = args_str.size();
        }
        else
        {
            controller->SetFailed( "serialize request error!" );
            LOG( ERROR ) << "serialize request error!";
        }

        // 就需要将服务名+方法名+参数长度封装到rpc头里面去
        RPC::RPCHeader rpcHeader;
        rpcHeader.set_method_name( service_name );
        rpcHeader.set_method_name( method_name );
        rpcHeader.set_args_size( args_size );

        // 接着就是将rpcHeader进行序列化
        std::string rpcHeader_srt;
        if ( !rpcHeader.SerializePartialToString( &rpcHeader_srt ) )
        {
            controller->SetFailed( "serialize rpc header error!" );
            LOG( ERROR ) << "rpcHeader serialize request error!";
        }

        // 首先定义一个需要发送的数据字符串
        std::string send_rpc_srt;
        {
            // 构建stringOutputStream流
            google::protobuf::io::StringOutputStream string_out_put( &send_rpc_srt );

            // 构建CodedOutputStream流
            google::protobuf::io::CodedOutputStream code_out_put( &string_out_put );

            // 将rpc_header的长度写入到send_rpc_srt中（变长编码）
            code_out_put.WriteVarint32( rpcHeader_srt.size() );

            // 将rpc_header写入到send_rpc_srt中
            code_out_put.WriteString( rpcHeader_srt );
        }

        // 附加消息的参数数据(即上面序列化的request字符串args_str)
        send_rpc_srt += args_str;

        // 发送数据,如果发送失败会重新连接，如果连接失败，则会返回错误信息，并return
        while ( send( m_client_fd, send_rpc_srt.c_str(), send_rpc_srt.size(), 0 ) == -1 )
        {
            perror( "send error!" );
            LOG( ERROR ) << "发送RPC数据出错，错误号为: " << errno;
            std::cout << "尝试重新连接，对方ip：" << m_ip << " 对方端口" << m_port << std::endl;
            LOG( INFO ) << "尝试重新连接，对方ip：" << m_ip << " 对方端口" << m_port;
            // 这里就是要重新连接了
            close( m_client_fd );
            m_client_fd = -1;
            std::string errMsg;
            bool ret = newConnect( m_ip.c_str(), m_port, errMsg );
            if ( !ret )
            {
                controller->SetFailed( errMsg );
                LOG( FATAL ) << errMsg;
            }
        }

        // 然后开始接收rpc处理结果
        std::vector< char > recv_buff( 1024 );
        int recv_size = 0;
        if ( ( recv_size = recv( m_client_fd, recv_buff.data(), recv_buff.size(), 0 ) ) == -1 )
        {
            close( m_client_fd );
            m_client_fd = -1;
            std::cerr << "从RPC服务端接收数据失败" << std::endl;
            LOG( ERROR ) << "从RPC服务端接收数据失败，错误号为: " << errno;
        }

        /**
         * @brief 然后这里就是处理接收到的数据了
         *      1、我们需要将接收到的数据反序列化成resopnse对象了
         */

        if ( !response->ParseFromArray( recv_buff.data(), recv_size ) )
        {
            close( m_client_fd );
            m_client_fd = -1;
            std::cerr << "反序列化RPC响应数据失败" << std::endl;
            LOG( ERROR ) << "反序列化RPC响应数据失败";
        }
    }

    bool MyrpcChannel::newConnect( const char *ip, uint16_t port, std::string &errMsg )
    {
        int clientFd = socket( AF_INET, SOCK_STREAM, 0 );
        if ( clientFd == -1 )
        {
            m_client_fd = -1;
            std::cerr << "create socket error!!,error: " << errno << std::endl;
            std::stringstream ss;
            ss << "create socket error!!,error: " << errno;
            errMsg = ss.str();
            LOG( ERROR ) << "create socket error!!,error: " << errno;
            return false;
        }

        struct sockaddr_in serverAddress;
        serverAddress.sin_family = AF_INET;
        serverAddress.sin_port = htons( port );
        inet_pton( AF_INET, ip, &serverAddress.sin_addr.s_addr );

        // 连接rpc节点
        if ( connect( clientFd, ( struct sockaddr * )&serverAddress, sizeof( serverAddress ) ) == -1 )
        {
            // 说明没有连接上
            m_client_fd = -1;
            std::cerr << "TCP connect is failed! error: " << errno << std::endl;
            LOG( FATAL ) << "TCP connect is failed! error: " << errno;
            std::stringstream ss;
            ss << "create socket error!!,error: " << errno;
            errMsg = ss.str();
            return false;
        }

        m_client_fd = clientFd;
        return true;
    }
}  // namespace RPC